// // producer.cpp

#include <librdkafka/rdkafkacpp.h>
#include <iostream>
#include <csignal>
#include <memory>


class ExampleDeliveryReportCb : public RdKafka::DeliveryReportCb {
public:
    void dr_cb(RdKafka::Message &message) override {
        if (message.err())
            std::cerr << "Message delivery failed: " << message.errstr() << std::endl;
        else
            std::cout << "Message delivered to topic " << message.topic_name()
                      << " [" << message.partition() << "] at offset "
                      << message.offset() << std::endl;
    }
};

class ExampleEventCb : public RdKafka::EventCb {
    void event_cb(RdKafka::Event &event) override {
        if (event.type() == RdKafka::Event::EVENT_ERROR) {
            std::cerr << "Kafka Error: " << event.str() << std::endl;
        } 
        std::cout << "Kafka Event: " << event.type() << " " << event.str() << std::endl;
    }
};

int main() {
    std::string brokers = "localhost:9092";
    std::string topic_str = "test_topic";
    std::string errstr;

    // 创建配置
    ExampleEventCb event_cb;
    std::unique_ptr<RdKafka::Conf> conf(RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL));
    conf->set("bootstrap.servers", brokers, errstr);
    conf->set("event_cb", &event_cb, errstr);
    ExampleDeliveryReportCb ex_dr_cb;
    conf->set("dr_cb", &ex_dr_cb, errstr);

    // 创建 producer
    std::unique_ptr<RdKafka::Producer> producer(RdKafka::Producer::create(conf.get(), errstr));
    if (!producer) {
        std::cerr << "Failed to create producer: " << errstr << std::endl;
        return 1;
    }

    // 创建 Topic
    std::unique_ptr<RdKafka::Topic> topic(RdKafka::Topic::create(producer.get(), topic_str, nullptr, errstr));
    if (!topic) {
        std::cerr << "Failed to create topic: " << errstr << std::endl;
        return 1;
    }

    std::string message = "Hello from C++ Kafka Producer!";
    RdKafka::ErrorCode resp = producer->produce(
        topic.get(),                            // topic ptr 目标Topic
        RdKafka::Topic::PARTITION_UA,           // partition       表示由 Kafka 自动分配
        RdKafka::Producer::RK_MSG_COPY,         // message flags   消息拷贝
        const_cast<char *>(message.c_str()),    // payload         消息内容
        message.size(),                         // payload size    消息长度
        nullptr,                                // optional key
        nullptr);                               // opaque

    if (resp != RdKafka::ERR_NO_ERROR) {
        std::cerr << "Produce failed: " << RdKafka::err2str(resp) << std::endl;
    } else {
        std::cout << "Message sent successfully\n";
    }

    producer->flush(3000);
    return 0;
}


